package com.jeffse.light.stream.core.handler;

import com.jeffse.light.stream.core.constans.CommandConstans;
import com.jeffse.light.stream.core.data.stream.DataStream;
import com.jeffse.light.stream.core.pojo.TranslateDataStream;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class StreamServerHander extends ChannelInboundHandlerAdapter {
    private Map<String, DataStream> dataStreamMap = new HashMap<>();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof TranslateDataStream) {
            TranslateDataStream translateDataStream = (TranslateDataStream) msg;
            switch (translateDataStream.getCommand()){
                case CommandConstans.START:
                    translateDataStream.getDataStream().start();
                    dataStreamMap.put(translateDataStream.getDataStreamName(), translateDataStream.getDataStream());
                    break;
                case CommandConstans.STOP:
                    if(dataStreamMap.containsKey(translateDataStream.getDataStreamName())){
                        dataStreamMap.get(translateDataStream.getDataStreamName()).stop();
                    }
                    break;
                case CommandConstans.UPDATE:
                    if(dataStreamMap.containsKey(translateDataStream.getDataStreamName())){
                        dataStreamMap.get(translateDataStream.getDataStreamName()).stop();
                    }
                    translateDataStream.getDataStream().start();
                    dataStreamMap.put(translateDataStream.getDataStreamName(), translateDataStream.getDataStream());
                    break;
                default:
                    throw new IllegalStateException("Unexpected value: " + translateDataStream.getCommand());
            }
        }else{
            log.error(msg.toString() + "");
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }

    public void cancel(){
//        ReferenceCountUtil.release(msg);
    }
}
